package com.peng.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/**
 * SpringBootKafkaProducer
 *
 * @author: lupeng6
 * @create: 2023/3/21 10:03
 */
@RestController
@RequestMapping("/kafka")
public class SpringBootKafkaProducerController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("sendMsg")
    public Map<String, Object> sendMsg(String msg) throws ExecutionException, InterruptedException {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("sample2", msg);
        Map<String, Object> result = new HashMap<>();
        result.put("partition", future.get().getRecordMetadata().partition());
        return result;
    }
}
